package com.wushi.bigdata;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.GenericInMemoryCatalog;
import org.apache.flink.table.catalog.hive.HiveCatalog;

import java.net.URL;

/**
 * @ClassName TableSql
 * 通过TableAPI执行相应的sql
 * @Deacription 通过定义不同的Catalog，可以在相应Catalog中创建源数据，并固化；避免多次创建； HiveCatalog依赖本地的hive配置：hive-site.xml,hdfs-site.xml
 * 默认Catalog为GenericMemoryCatalog内存型，所以需要每次都创建一遍；
 *
 * 时态表（temporal table） Join
 * 仅支持INNER JOIN与LEFT JOIN。在join的时候需要使用 FOR SYSTEM_TIME AS OF  table1.proctime as XXX，
 * 其中table1.proctime表示table1的proctime处理时间属性(计算列)，只匹配当前处理时间维表所对应的的快照数据。
 * 目前不支持基于事件时间(event time)的temporal table join
 * 维表可能会不断变化，JOIN行为发生后，维表中的数据发生了变化（新增、更新或删除），则已关联的维表数据不会被同步变化
 * 维表和维表不能进行JOIN
 * 维表必须指定主键,维表JOIN时，ON的条件必须包含所有主键的等值条件
 *
 * // 如果不加FOR SYSTEM_TIME AS OF, 则作为静态宝进行join，join完后task关闭，无法支持Checkpoint!!!!
 *
 * @Author wushumin
 * @Date 2021-08-03 10:25
 * @Version 1.0
 **/
public class TemporalTableSqlJoinCanalJsonAgentInidicator {
    public StreamTableEnvironment getEnv(){
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        environment.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
        //指定checkpoint时间间隔，并指定checkpoint的模式，是exactly-once（刚好一次）还是AT_LEAST_ONCE（至少一次）。
        // 大多数情况下是exactly-once（默认就是这个模式），少数情况下，如果要求超低延迟的处理情况，才会设置AT_LEAST_ONCE
        //    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    /* -------------设置checkpoint上一个的结束点到下一个开始点之间的最短时间。
    因为checkpoint触发时，需要一定时间去完成整个checkpoint的过程，
    如果checkpoint的完成时间过程，导致前后两个checkpoint间的时间间隔过短，这是不合适的，没有必要。
    1、这里的时间间隔，指的是上一个checkpoint完成的时间点，到下一个checkpoint开始的时间点的间隔，如果过短，会导致频繁checkpoint，影响性能。假设这个间隔为T
    2、而上面设置的checkpoint时间间隔，指的是前一个checkpoint的开始时间到下一个checkpoint的开始时间。所以是始终大于1中的时间间隔的。假设这个间隔为 N

    如果T小于这里设置的值，那么无论N设置多少，下一个checkpoint的开始时间必须是500ms之后。
    如果T大于这里设置的值，那么正常按照N设置的间隔来触发下一个checkpoint，这里设置的间隔无关了。*/
        environment.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        //    // 设置checkpoint完成的超时时间
        environment.getCheckpointConfig().setCheckpointTimeout(60000);
        //    // 设置checkpoint的最大并行度
        environment.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //      开启checkpoints的外部持久化，但是在job失败的时候不会自动清理，需要自己手工清理state
        //     DELETE_ON_CANCELLATION:在job canceled的时候会自动删除外部的dwd_fact_receive_status状态数据，但是如果是FAILED的状态则会保留；
        //      RETAIN_ON_CANCELLATION:在job canceled的时候会保留状态数据
        environment.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//    val dir ="hdfs://nameservice1/user/flink/checkpoint/"
//    val backend:StateBackend = new RocksDBStateBackend(dir)
        environment.setStateBackend(new EmbeddedRocksDBStateBackend());
//    environment.setStateBackend(backend)
//    val dir =ConfigMgtV2.loadApolloPropery("hadoop.name.service")

        String dir ="file:///D:/hdfs/";
        environment.getCheckpointConfig().setCheckpointStorage(dir);


        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment,environmentSettings);
        URL resources = Thread.currentThread().getContextClassLoader().getResource(".");
        String path = resources.getPath();
        HiveCatalog hiveCatalog = new HiveCatalog("hiveCat", "lcb_db", path);
        tableEnvironment.registerCatalog("hiveCat",hiveCatalog);
        GenericInMemoryCatalog memoryCat = new GenericInMemoryCatalog("memoryCat");
        tableEnvironment.registerCatalog("memoryCat",memoryCat);
        return tableEnvironment;
    }

    public void process(StreamTableEnvironment env){
        env.executeSql("create database if not exists `lcb_mic_market_db`");
        env.executeSql("create database if not exists `lcb_mars_micro_ads_db`");

//        注册source流表
        env.executeSql("CREATE TABLE IF NOT EXISTS `lcb_mic_market_db`.`mic_agent_user`(\n" +
                " `id` BIGINT  COMMENT '主键',\n" +
                "`store_id` INT  COMMENT '店铺id',\n" +
                "`agent_user_id` BIGINT  COMMENT '代理人micro_user_id',\n" +
                "`agent_user_mobile` VARCHAR  COMMENT '代理人手机号',\n" +
                "`agent_user_name` VARCHAR  COMMENT '代理人名称',\n" +
                "`agent_status` TINYINT   COMMENT '代理人状态，1正常，2拉黑，3冻结，默认1',\n" +
                "`bind_user_count` INT  COMMENT '绑定客户数',\n" +
                "`order_count` INT  COMMENT '代理订单数',\n" +
                "`order_total_amount` INT  COMMENT '已获得佣金金额（分）',\n" +
                "`withdraw_amount` INT  COMMENT '已提现金额（分）',\n" +
                "`is_store_add` TINYINT  COMMENT '是否店铺添加，0否，1是，默认0',\n" +
                "`operator_id` BIGINT  COMMENT '操作人id，后台操作时不为空',\n" +
                "`operator_name` VARCHAR  COMMENT '操作人姓名',\n" +
                "`created_time` BIGINT  COMMENT '创建时间',\n" +
                "`updated_time` BIGINT  COMMENT '修改时间',\n" +
                "  proctime as PROCTIME()" +     //取处理事件，作为虚拟列！！！
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'flink_sql_test',\n" +
                "  'properties.bootstrap.servers' = 'cdh1.lcbint.cn:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' =  'latest-offset',\n" +

                " 'canal-json.database.include'='lcb_mic_market_db',\n" +  //只获取canal-json中指定database的数据
                " 'canal-json.table.include'='mic_agent_user',\n" +  //只获取canal-json中指定table的数据， 可通过此规则对数据进行校验
                "  'format' = 'canal-json'\n" +
                ")");

//        注册IMpala维度表
        String Mic_user_dim_sql=""+
//                "CREATE TABLE us_micro_user (\n" +
//                "  `id`  bigint ,\n" +
//                "  `user_id` bigint,  \n" +
//                "  `group_id` int,  \n" +
//                "  `openid` varchar,  \n" +
//                "  `real_name` varchar  \n" +
//                ")WITH (\n" +
//                "  'connector.type' = 'jdbc',\n" +
//                "  'connector.url' = 'jdbc:impala://192.168.6.76:21050/lcb_mic_user_db;UseNativeQuery=1',\n" +
//                "  'connector.table' = 'us_micro_user',\n" +
//              "  'connector.driver' = 'com.cloudera.impala.jdbc41.Driver',\n" +
//                "  'connector.lookup.cache.max-rows' = '100', -- 缓存条数 \n"+
//                "  'connector.lookup.cache.ttl' = '30s' -- 缓存时间 \n"+
//                ")";
//        env.executeSql(Mic_user_dim_sql);
//        注册sink流表
        env.executeSql("CREATE TABLE if not EXISTS `lcb_mars_micro_ads_db`.`micro_indicator_agent_nums`(\n" +
                "id bigint,\n" +
                "indicator_code string,\n" +
                "metric_code string,\n" +
                "dimension_code string,\n" +
                "indicator_date bigint,\n" +
                "indicator_cycle string,\n" +
                "group_id bigint,\n" +
                "group_type int,\n" +
                "`value` bigint,\n" +
                "created_time bigint,\n" +
                "updated_time bigint,\n" +
                "PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = 'flink_sql_sink_test',\n" +
                "  'properties.bootstrap.servers' = 'cdh1.lcbint.cn:9092',\n" +
                  //TODO KEY 与source的不同！！
                " 'value.canal-json.database.include'='flink_sink_canal_db',\n" +  //只获取canal-json中指定database的数据
                " 'value.canal-json.table.include'='micro_indicator_agent_nums',\n" +  //只获取canal-json中指定table的数据， 可通过此规则对数据进行校验

                "  'key.format' = 'json',\n" +
                "  'value.format' = 'canal-json'\n" +   //sink采用canal-json格式依赖 flink-json-lecb-ext-1.13.1.jar
                ")");


//                env.executeSql("select * from `lcb_mic_market_db`.`mic_agent_user`").print();
// 插入kafka sink流表
//        TableResult result = env.executeSql("insert into KafkaSinkTable select user_id,item_id, behavior from KafkaTable where user_id %2=0");

        //需要使用 SYSTEM_TIME AS OF 来join时态表 us_micro_user ！！！
        // 如果不加在则作为静态宝进行join，join完后task关闭，无法支持Checkpoint
        env.executeSql(" INSERT INTO\n" +
                "            `lcb_mars_micro_ads_db`.`micro_indicator_agent_nums`\n" +
                "            ( id,\n" +
                "            indicator_code,\n" +
                "            metric_code,\n" +
                "            dimension_code,\n" +
                "            indicator_date,\n" +
                "            indicator_cycle,\n" +
                "            group_id,\n" +
                "            group_type,\n" +
                "            `value`,\n" +
                "            created_time,\n" +
                "            updated_time\n" +
                "            )\n" +
                "       select " +
                        "10001,"+
                        "'agent_indicator_code1',\n"+
                "        'metric_code1',\n" +
                "        'dimension_code1',\n" +
                "         1592446973647,\n" +
                "         'day',\n" +
                "        store_id,\n" +
                "         2,\n"+
                "        count(1),\n" +
                "       1629367870997,\n"+
                "       1629367870997\n"+
                "        from `lcb_mic_market_db`.`mic_agent_user`\n" +
                "        where \n" +
                "        store_id is not null\n" +
                "        group by store_id");
//
//        env.createTemporaryView("join_table", jointTable);
//        TableResult result = env.executeSql("insert into flink_kafka_sink select * from join_table");
//        env.executeSql("select * from `lcb_mars_micro_ads_db`.`micro_indicator_agent_nums`").print();

    }

    public static void main(String[] args) {
        TemporalTableSqlJoinCanalJsonAgentInidicator tableSql = new TemporalTableSqlJoinCanalJsonAgentInidicator();
        StreamTableEnvironment env = tableSql.getEnv();
        tableSql.process(env);
    }
}

